package algorithms.bloom;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Collections;
import storage.Block;
import configuration.Configuration;
import simulator.Client;
import simulator.Manager;

/**
 * Client node that works under bloom filter-based algorithm.
 * 
 * @author Nodirjon Siddikov
 */
public class BloomClient extends Client {
	// local cache
	// key - cache block ID (different than block ID)
	// value - cache block content
	private HashMap<Integer, LocalCacheBlock> _lclCache;
	// global cache
	// key - remote client ID
	// value - remote cache information
	private HashMap<Integer, BloomFilter<String>> _gblCache; // grouped by
																// remote client
																// ID
	private BloomManager _manager;

	/**
	 * Constructor
	 * 
	 * @param id
	 *            ID of the client
	 */
	public BloomClient(int id) {
		super(id); // call to the constructor of the super class
		_lclCacheSize = Configuration.localCacheSize; // in Bytes
		// calculate the max number of cache blocks
		// that local cache memory can accommodate
		_maxNoOfCacheBlocks = _lclCacheSize / Configuration.localCacheBlockSize;
		// calculate the initial hashtable capacity
		int initialCapacity = (int) Math.ceil(_maxNoOfCacheBlocks
				/ _hashTableLoadFactor) + 1;
		_lclCache = new HashMap<Integer, LocalCacheBlock>(initialCapacity,
				_hashTableLoadFactor);
		_gblCache = new HashMap<Integer, BloomFilter<String>>();
		if (Configuration.detailedStats != 0) {
			System.out.println("Bloom client " + getID() + " is up.");
			System.out
					.println("Local cache size: " + _lclCacheSize + " Bytes.");
			System.out.println("Max no. of cache blocks: "
					+ _maxNoOfCacheBlocks);
		}
	}

	/**
	 * This method sets the manager to the clients
	 * 
	 * @param m
	 *            The manager
	 */
	public void setManager(Manager m) {
		_manager = (BloomManager) m;
	}

	/**
	 * This method is used to broadcast the local cache summary to the clients.
	 */
	private void broadcastCacheSummary() {
		// get the list of clients from the manager
		BloomClient[] clients = _manager.getListOfClients(this.getID());
		// generate the cache summary
		BloomFilter<String> bf = this.generateCacheSummary();
		if (Configuration.detailedStats != 0) {
			System.out.println();
			System.out.println();
			System.out
					.println("Client "
							+ this.getID()
							+ " is broadcasting digest.----------------------------------------------------->");
		}
		// broadcast the cache summary to all clients
		for (BloomClient c : clients) {
			c.receiveCacheSummary(this.getID(), bf);
		}
		_digestExchangeCount++;
	}

	/**
	 * Compute the bloom filter representation of the local cache
	 * 
	 * @return A bloom filter
	 */
	private BloomFilter<String> generateCacheSummary() {
		BloomFilter<String> bf = null;
		// the following variable defines how to initilize the bloom filter
		int constOption = Configuration.bloomFilterConstructor;
		switch (constOption) {
		case 0: // 0 - set m, n and k
		{
			int n = _maxNoOfCacheBlocks;
			int m = n * Configuration.bloomFilterLoadFactor; // calculate bitset
																// size
			int k = Configuration.noOfHashFunctions;
			bf = new BloomFilter<String>(m, n, k);
			break;
		}
		case 1: // 1 - set m and n, an optimal k is calculated automatically
		{
			int n = _maxNoOfCacheBlocks;
			int m = n * Configuration.bloomFilterLoadFactor; // calculate bitset
																// size
			bf = new BloomFilter<String>(m, n);
			break;
		}
		}
		// calculate the bloom filter
		for (LocalCacheBlock e : _lclCache.values()) {
			// don't include dead blocks into bloom filter
			if (e.isAlive()) {
				// add to filter (disk block ID and its originality info)
				bf.add(e.getDiskBlockID() + "_" + e.isOriginal());
			}
		}
		// return the computed bloom filter
		return bf;
	}

	/**
	 * Receive the cache summary from the remote client and update the global
	 * cache with this cache summary
	 * 
	 * @param rmtClientID
	 *            the ID of the remote client
	 * @param rmtCacheSummary
	 *            the summary of the remote cache
	 */
	private void receiveCacheSummary(int rmtClientID,
			BloomFilter<String> rmtCacheSummary) {
		// update the global cache with remote cache summary
		_gblCache.put(rmtClientID, rmtCacheSummary);
		_digestExchangeCount++;
		// resolve a duplicate original blocks
		boolean isOriginal = true;
		for (LocalCacheBlock b : _lclCache.values()) {
			// if there is an original remote block such that
			// its contents are same as an original local block
			if (rmtCacheSummary.contains(b.getDiskBlockID() + "_" + isOriginal)) {
				// mark local original block as non-original
				b.setOriginal(false);
			}
		}
		if (Configuration.detailedStats != 0) {
			printLocalCacheContent();
		}
	}

	// receive a read request
	public Block receiveReadRequest(int diskBlockID) {
		// periodically broadcast digest
		if (_ttlNoOfReadRequests % Configuration.broadcastInterval == 0) {
			broadcastCacheSummary();
		}
		if (Configuration.detailedStats != 0) {
			System.out.println();
			System.out.println("Client " + this.getID()
					+ " is requesting block " + diskBlockID);
		}
		// STEP
		// 1:////////////////////////////////////////////////////////////////////////////////
		_blckAccessTime = 0; // reset the block access time
		decrementClocks(); // decrement clocks
		_ttlNoOfReadRequests++; // increment the number of read requests
								// received
		LocalCacheBlock lclCacheBlock = null;
		// search local cache memory
		for (Map.Entry<Integer, LocalCacheBlock> b : _lclCache.entrySet()) {
			// if found
			if (b.getValue().getDiskBlockID() == diskBlockID) {
				lclCacheBlock = _lclCache.get(b.getKey());
				break;
			}
		}
		_blckAccessTime += Configuration.cacheAccessTime;
		// if the block is cached locally
		if (lclCacheBlock != null) {
			_lclCacheHitCount++; // increment local cache hit
			_ttlBlockAccessTime += _blckAccessTime;
			if (Configuration.detailedStats != 0) {
				System.out.println("Local cache hit.");
				System.out.println("Block access time: " + _blckAccessTime);
			}
			lclCacheBlock.resetClock(); // reset the block's clock count to zero
			return lclCacheBlock.getDiskBlock(); // serve the disk block
		}
		// STEP
		// 2:////////////////////////////////////////////////////////////////////////////////
		_lclCacheMissCount++; // increment the local cache miss
		if (Configuration.detailedStats != 0) {
			System.out.println("Local cache miss.");
		}
		int rmtClientID = -1;
		// search the global cache memory
		for (int clientID : _gblCache.keySet()) {
			BloomFilter<String> bf = _gblCache.get(clientID);
			// search for an ORIGINAL block
			boolean isOriginal = true;
			if (bf.contains(diskBlockID + "_" + isOriginal)) {
				// get the remote client ID
				rmtClientID = clientID;
				break; // the for loop
			}
			// search for an NON-ORIGINAL block
			isOriginal = false;
			if (bf.contains(diskBlockID + "_" + isOriginal)) {
				// get the remote client ID
				rmtClientID = clientID;
				break; // the for loop
			}
		}
		_blckAccessTime += Configuration.cacheAccessTime;
		BloomClient rmtClient = _manager.getClient(rmtClientID);
		Block rmtDiskBlock = null;
		if (rmtClient != null) {
			if (Configuration.detailedStats != 0) {
				System.out.println("Client " + rmtClientID
						+ " has the desired disk block " + diskBlockID);
			}
			// fetch the disk block
			rmtDiskBlock = rmtClient.serveDiskBlock(diskBlockID);
			_blckAccessTime += (Configuration.messageLatency + Configuration.cacheAccessTime);
			if (rmtDiskBlock != null) {
				// increment global cache hit
				_rmtCacheHitCount++;
				_blockReadCount++;
				_ttlBlockAccessTime += _blckAccessTime;
				if (Configuration.detailedStats != 0) {
					System.out.println("Remote cache hit.");
					System.out.println("Disk block is fetched from client "
							+ rmtClient.getID());
					System.out.println("Block access time: " + _blckAccessTime);
				}
				boolean isOriginal = false;
				cacheDiskBlock(null, rmtDiskBlock, isOriginal);
				return rmtDiskBlock;
			}
		}
		// STEP
		// 3:////////////////////////////////////////////////////////////////////////////////
		// increment remote cache miss
		_rmtCacheMissCount++;
		if (Configuration.detailedStats != 0) {
			System.out.println("Remote cache miss.");
		}
		// try fetching the block from server
		rmtDiskBlock = _manager.readFromStorage(diskBlockID);
		_lookUpCount++;
		_blckAccessTime += (2 * Configuration.messageLatency + Configuration.storageAccessTime); // client
																									// ->
																									// mgr
																									// ->
																									// storage
		if (rmtDiskBlock != null) {
			_serverDiskHitCount++;
			_ttlBlockAccessTime += _blckAccessTime;
			if (Configuration.detailedStats != 0) {
				System.out.println("Block access time: " + _blckAccessTime);
			}
			boolean isOriginal = true;
			cacheDiskBlock(null, rmtDiskBlock, isOriginal);
			return rmtDiskBlock;
		}
		return null;
	}

	// decrement clocks of local cache blocks
	private void decrementClocks() {
		for (LocalCacheBlock entry : _lclCache.values()) {
			entry.decrementClock();
		}
	}

	// serve the disk block from remote client
	private Block serveDiskBlock(int diskBlockID) {
		// read the disk block from local cache memory
		for (LocalCacheBlock b : _lclCache.values()) {
			if (b.getDiskBlockID() == diskBlockID) {
				return b.getDiskBlock();
			}
		}
		return null;
	}

	// cache the disk block
	private void cacheDiskBlock(ArrayList<Integer> fwdingClients, Block block,
			boolean isOriginal) {
		// check if eviction procedure is needed
		// compare the current size against the maximum no of cache blocks
		if (_lclCache.size() >= this._maxNoOfCacheBlocks) {
			evictBlock(fwdingClients);
		}
		// create a new disk block (with new memory address in RAM)
		LocalCacheBlock lclCacheBlock = new LocalCacheBlock(new Block(
				block.getID(), block.getSize()), isOriginal);
		// get the maximum cache block ID
		int newCacheBlockID = 0;
		if (!_lclCache.isEmpty()) {
			newCacheBlockID = Collections.max(_lclCache.keySet()) + 1;
		}
		// add new cache block to the LCI
		_lclCache.put(newCacheBlockID, lclCacheBlock);
		if (Configuration.detailedStats != 0) {
			System.out.println("Client " + this.getID()
					+ " locally cached the disk block " + block.getID());
			printLocalCacheContent();
		}
	}

	private void printLocalCacheContent() {
		System.out.println("Client " + this.getID()
				+ "'s local cache content: ");
		System.out.print("Alive: ");
		for (LocalCacheBlock e : _lclCache.values()) {
			String original = (e.isOriginal()) ? "r" : "";
			if (e.isAlive()) {
				System.out.print(e.getDiskBlockID() + original + " ");
			}
		}
		System.out.println();
		System.out.print("Dead: ");
		for (LocalCacheBlock e : _lclCache.values()) {
			String original = (e.isOriginal()) ? "r" : "";
			if (!e.isAlive()) {
				System.out.print(e.getDiskBlockID() + original + " ");
			}
		}
		System.out.println();
	}

	// evict the cache block from local cache memory
	private void evictBlock(ArrayList<Integer> fwdingClients) {
		// search for a dead block
		for (Map.Entry<Integer, LocalCacheBlock> e : _lclCache.entrySet()) {
			// look for DEAD blocks
			if (!e.getValue().isAlive()) {
				// evict the cache blocks
				_lclCache.remove(e.getKey());
				if (Configuration.detailedStats != 0) {
					System.out.println("Dead disk block "
							+ e.getValue().getDiskBlockID() + " is evicted.");
				}
				return;
			}
		}
		// if no DEAD cache block is found
		// evict the oldest non-original block
		int oldestNonOrigClock = 0; // get the first local cache block
		int oldestNonOrigCacheBlockID = -1; // cache block ID
		boolean isOldestNonOriginalFound = false;
		// find oldest block
		for (Map.Entry<Integer, LocalCacheBlock> e : _lclCache.entrySet()) {
			if (!e.getValue().isOriginal()) {
				// if the block is oldest and non-original
				if (e.getValue().getClock() <= oldestNonOrigClock) {
					oldestNonOrigClock = e.getValue().getClock();
					oldestNonOrigCacheBlockID = e.getKey();
					isOldestNonOriginalFound = true;
				}
			}
		}
		// evict the oldest non-original block
		if (isOldestNonOriginalFound) {
			_lclCache.remove(oldestNonOrigCacheBlockID); // delete based on
															// cache block ID
			if (Configuration.detailedStats != 0) {
				System.out
						.println("Oldest NON-ORIGINAL disk block is removed from client "
								+ this.getID());
			}
			return;
		}
		// if non-original block is NOT found, then evict the oldest original
		// block
		int oldestOrigClock = 0;
		int oldestOrigCacheBlockID = -1; // cache block ID
		int diskBlockID = -1;
		// find oldest block
		for (Map.Entry<Integer, LocalCacheBlock> e : _lclCache.entrySet()) {
			// if the block is oldest and non-original
			if (e.getValue().getClock() <= oldestOrigClock) {
				oldestOrigClock = e.getValue().getClock();
				oldestOrigCacheBlockID = e.getKey(); // cache block ID
				diskBlockID = e.getValue().getDiskBlockID();
			}
		}
		// check if other peer already has a non-original block
		boolean isMarked = false;
		// check if other peer already has a non-original block
		for (int rmtClientID : _gblCache.keySet()) {
			BloomFilter<String> bf = _gblCache.get(rmtClientID);
			boolean isOriginal = false;
			// check if remote entry exists with same disk block ID
			if (bf.contains(diskBlockID + "_" + isOriginal)) {
				// get the remote client ID
				BloomClient rmtClient = _manager.getClient(rmtClientID);
				// try to mark the block
				isMarked = rmtClient.markAsOriginal(diskBlockID);
				break; // the for loop
			}
		}
		if (!isMarked) {
			// create a list of forwarding clients
			if (fwdingClients == null) {
				fwdingClients = new ArrayList<Integer>();
			}
			fwdingClients.add(this.getID());
			relocate(fwdingClients, _lclCache.get(oldestOrigCacheBlockID)
					.getDiskBlock());
		}
		// remove the original block from current client's cache memory
		_lclCache.remove(oldestOrigCacheBlockID); // cache block ID
		if (Configuration.detailedStats != 0) {
			System.out.println("Oldest ORINIGAL disk block " + diskBlockID
					+ " is removed from client " + this.getID());
		}
	}

	// mark the local cache block as original
	private boolean markAsOriginal(int diskBlockID) {
		for (LocalCacheBlock e : _lclCache.values()) {
			if (e.getDiskBlockID() == diskBlockID) {
				e.setOriginal(true);
				if (Configuration.detailedStats != 0) {
					System.out.println("Disk block " + diskBlockID
							+ " is marked as original.");
				}
				return true; // marking is successful
			}
		}
		return false; // marking is unsuccessful
	}

	// relocate the cache block to one of the idle caches
	private void relocate(ArrayList<Integer> fwdingClients, Block diskBlock) {
		// find a potential collaborator peer
		// by computing the corresponding number of free cache blocks in each
		// peer
		BloomClient[] clients = _manager.getListOfClients(this.getID());
		int minOrignialBlocks = clients[0].calculateNoOfOriginalCacheBlocks();
		BloomClient ptnCollaborator = null;
		// find a potential collaborator
		for (BloomClient c : clients) {
			// exclude the current client
			if (fwdingClients != null && fwdingClients.contains(c.getID())) {
				continue;
			} else {
				if (c.calculateNoOfOriginalCacheBlocks() <= minOrignialBlocks) {
					minOrignialBlocks = c.calculateNoOfOriginalCacheBlocks();
					ptnCollaborator = c;
				}
			}
		}
		// send a store request for the favorite peer
		if (ptnCollaborator != null) {
			if (Configuration.detailedStats != 0) {
				System.out.println("Potential collaborator is found: client "
						+ ptnCollaborator.getID());
				System.out.println("Original disk block " + diskBlock.getID()
						+ " is being relocated " + ptnCollaborator.getID());
			}
			ptnCollaborator.storeRequest(fwdingClients, diskBlock);
			_relocationCount++;
		} else {
			if (Configuration.detailedStats != 0) {
				System.out.println("Potential collaborator is NOT found.");
				System.out.println("Original disk block " + diskBlock.getID()
						+ " is being discarded compeletely.");
			}
		}
	}

	// store the forwarded disk block
	public void storeRequest(ArrayList<Integer> fwdingClients, Block diskBlock) {
		_relocationCount++;
		// search the local cache memory for particular disk block
		for (LocalCacheBlock b : _lclCache.values()) {
			// if found
			if (b.getDiskBlockID() == diskBlock.getID()) {
				b.setOriginal(true);
				return;
			}
		}
		boolean isOriginal = true;
		this.cacheDiskBlock(fwdingClients, diskBlock, isOriginal);
	}

	// calculate the number of original blocks
	private int calculateNoOfOriginalCacheBlocks() {
		int origCacheBlockCount = 0;
		for (LocalCacheBlock e : _lclCache.values()) {
			if (e.isOriginal()) {
				origCacheBlockCount++;
			}
		}
		return origCacheBlockCount;
	}

	// calculate the size of the coop cache size
	public int getClientCoopCacheSize() {
		// in bytes
		// we have no direct reference to the bloom filter,
		// so we cannot use size() method here to get m
		// thus we calculate m manually
		int bloomFilterSize = (_maxNoOfCacheBlocks * Configuration.bloomFilterLoadFactor) / 8;
		int noOfGblCacheBlocks = _gblCache.size();
		return bloomFilterSize * noOfGblCacheBlocks;
	}

	// used for calculation of communication overhead for client (in and out of
	// Bytes)
	public double getDigestExchangeFraction() {
		double bloomFilterSize = (double) (_maxNoOfCacheBlocks * Configuration.bloomFilterLoadFactor) / 8;
		return _digestExchangeCount * bloomFilterSize;
	}

	public double getBlockReadFraction() {
		return _blockReadCount * Configuration.blockSize;
	}

	public double getRelocationFraction() {
		return _relocationCount * Configuration.blockSize;
	}

	public double getHintExchangeFraction() {
		return 0;
	}
}